package hit.edu.zjc.parsers;

import java.io.DataOutput;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.ByteBuffer;
import java.util.Arrays;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParser;
import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;

/*
 * 只适用于纯二进制（byte）的读
 * */
public class ByteTupleParserFactory implements ITupleParserFactory {
    private static final long serialVersionUID = 1L;


    public ByteTupleParserFactory() {

    }

    @Override
    public ITupleParser createTupleParser(final IHyracksTaskContext ctx) {
        return new ITupleParser() {
            @Override
            public void parse(InputStream in, IFrameWriter writer) throws HyracksDataException {
		        	
            	try{
		            	ByteBuffer frame = ctx.allocateFrame();
		            	frame.clear();
		            	byte[] data=new byte[1024];
		            	int len=in.read(data);
		            	while(len>0)
		            	{
		            		int remain=frame.remaining();
		            		if(len<=remain)
		            			frame.put(data, 0, len);
		            		else//frame将会装满
		            		{
		            			frame.put(data,0,remain);
		            			FrameUtils.flushFrame(frame, writer);
		            			frame.clear();
		            			frame.put(data,remain,len-remain);
		            		}
		            		len=in.read(data);
		            	}
		            	if(frame.remaining()!=frame.capacity())
		            		FrameUtils.flushFrame(frame, writer);
		            		
		            }
		        	catch(Exception e)
		        	{
		        		e.printStackTrace();
		        	}
          }
            
            	
        };
    }
}